package com.java.base.exercise.iostream;

import lombok.SneakyThrows;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author RenPu
 * @Date 2023/3/8 14:52
 * @Version 1.0
 * @Description: NIO结合多路复用Reactor模型 实现
 **/
public class NIOServer3 {

    //创建业务处理线程池
    private static ExecutorService workPoool = Executors.newCachedThreadPool();

    private ServerSocketChannel serverSocketChannel;


    //封装Reactor线程，进行处理selector的轮询监听事件，例如：accpect，io读写事件
    abstract class ReactorThread extends Thread {

        //轮询器对象
        Selector selector;

        //任务队列
        LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();

        //selector监听到所有事件后，就会调用这个方法
        public abstract void handler(SelectableChannel channel) throws Exception;

        private ReactorThread() throws Exception {
            selector = Selector.open();
        }

        volatile boolean running = false;

        @Override
        public void run() {
            while (running) {

                try {
                    Runnable task;
                    //不断从任务队列taskQueue进行获取任务，并进行执行对应的任务
                    while ((task = taskQueue.poll()) != null) {
                        task.run();
                    }

                    //超时时间设置
                    selector.select(1000);

                    //获取selector中的查询结果
                    Set<SelectionKey> selected = selector.selectedKeys();

                    //进行迭代遍历处理
                    Iterator<SelectionKey> iterator = selected.iterator();
                    while (iterator.hasNext()) {
                        SelectionKey key = iterator.next();
                        iterator.remove();
                        int ops = key.readyOps();

                        //设置关注的事件类型
                        if ((ops & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || ops == 0) {
                            SelectableChannel channel = (SelectableChannel) key.attachment();
                            channel.configureBlocking(false);
                            handler(channel);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }


            }
        }


        //处理链接后的客户端具体IO事件
        protected abstract void doStart();

        //注册Channel
        protected abstract SelectionKey register(SelectableChannel sch);
    }


    //创建处理客户端accpect的线程池
    private ReactorThread[] mainReactorThread = new ReactorThread[1];

    //创建处理客户端IO事件的线程池(订阅IO事件发生的线程)
    private ReactorThread[] subReactorThread = new ReactorThread[8];


    /**
     * 初始化线程组
     */
    @SneakyThrows
    private void newGroup() {


        //创建IO线程组，主要用于处理客户端连接后，针对SocketChannel的事件进行监听和处理 例如：内容的写入，读取
        for (int i = 0; i < subReactorThread.length; i++) {
            subReactorThread[i] = new ReactorThread() {
                @Override
                public void handler(SelectableChannel channel) throws Exception {

                    //将对象强转为SocketChannel
                    SocketChannel ch = (SocketChannel) channel;

                    //创建bytebuf作为获取内容的载体
                    ByteBuffer buffer = ByteBuffer.allocate(1024);

                    //判断通道的内容是否读取完毕，读取完毕则进行返回
                    while (ch.isOpen() && ch.read(buffer) != -1) {
                        //长链接下，需要手动判断数据有没有进行读取结束（此处做一个判断：超过0字节就认为请求结束了）
                        if (buffer.position() > 0) break;
                    }

                    //如果没有数据，则不进行下面逻辑的处理
                    if (buffer.position() == 0) return;

                    //由向buffer中写入数据，变为从buffer获取数据
                    buffer.flip();

                    //根据需要读取内容的长度，进行初始化储存数组的大小
                    byte[] content = new byte[buffer.limit()];

                    //buffer向数组储存对象中写入数据
                    buffer.get(content);

                    //TODO 进行根据具体的业务逻辑进行设计，例如：数据库持久化，接口
                    workPoool.submit(() -> {

                    });

                    //响应结果 200

                    String reponsed = "HTTP/1.1 200 OK\r\n" + "Content-Length: 11\r\n\r\n" + "Hello World !!!!";

                    ByteBuffer byteBuffer = ByteBuffer.wrap(reponsed.getBytes());

                    while (buffer.hasRemaining()) {
                        ch.write(byteBuffer);
                    }
                }

                @Override
                protected void doStart() {

                }

                @Override
                protected SelectionKey register(SelectableChannel sch) {
                    return null;
                }
            };

        }

        //创建线程组，主要用于处理客户端连接进行时accpect事件进行分发的处理，不做具体请求的处理
        for (int i = 0; i < mainReactorThread.length; i++) {

            AtomicInteger incr = new AtomicInteger(0);
            //获取socketChannel,监听accpect方法，同时设置为非阻塞的状态
            mainReactorThread[i] = new ReactorThread() {
                @Override
                public void handler(SelectableChannel channel) throws Exception {

                    ServerSocketChannel serverChannel = (ServerSocketChannel) channel;
                    SocketChannel sch = serverChannel.accept();

                    //每次都要进行设置的原因，就是系统操作默认都是阻塞IO模型
                    sch.configureBlocking(false);

                    //收到建立的链接，需要从IO处理线程组，随机选择一个线程进行处理
                    int index = incr.getAndIncrement() % subReactorThread.length;
                    ReactorThread workEventLoop = subReactorThread[index];
                    workEventLoop.doStart();

                    //注册Channel
                    SelectionKey selectionKey = workEventLoop.register(sch);
                    selectionKey.interestOps(SelectionKey.OP_READ);
                    System.out.println(Thread.currentThread().getName() + "收到新连接：" + sch.getRemoteAddress());

                }

                @Override
                protected void doStart() {

                }

                @Override
                protected SelectionKey register(SelectableChannel sch) {
                    return null;
                }


            };


        }

    }


    /**
     * 初始化channel,并进行绑定一个eventLoop线程组
     */
    @SneakyThrows
    public void initAndRegister(){

        //创建ServerSocketChannel对象
         serverSocketChannel = ServerSocketChannel.open();

        serverSocketChannel.configureBlocking(false);
        //从mainReactThread线程组获取一个线程进行处理
        int index=new Random().nextInt()%mainReactorThread.length;
        mainReactorThread[index].doStart();
        //将channel注册到selector
        SelectionKey  selectionKey= mainReactorThread[index].register(serverSocketChannel);
        selectionKey.interestOps(SelectionKey.OP_READ);

    }


    /**
     * 绑定端口的方法
     */

    @SneakyThrows
    public void bindPort(){
        serverSocketChannel.bind(new InetSocketAddress(8080));
        System.out.println("端口绑定成功。。。。。。。");
    }


    public static void main(String[] args) {
        NIOServer3 nioServer3 = new NIOServer3();
        nioServer3.newGroup();   //创建sub和main线程组
        nioServer3.initAndRegister();   //创建serverChannel,注册到main线程上的selector
        nioServer3.bindPort();          //绑定端口

    }





}
